package com.atguigu.gmall.mq.service;

import com.atguigu.gmall.common.constant.RedisConst;
import com.atguigu.gmall.common.util.JSONs;
import com.atguigu.gmall.common.util.MD5;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * @author lfy
 * @Description
 * @create 2023-08-04 11:25
 */
@Slf4j
@Service
public class RabbitMqService {

    private RabbitTemplate rabbitTemplate;
    private StringRedisTemplate redisTemplate;

    /**
     * 构造器注入：如果只有一个有参构造器，构造器中的所有参数，都会从容器中获取
     * 优点：强制性要求依赖的组件必须存在，如果new这个组件，也必须自己传
     * @param rabbitTemplate
     */
    public RabbitMqService(RabbitTemplate rabbitTemplate, StringRedisTemplate redisTemplate){
        this.rabbitTemplate = rabbitTemplate;
        this.redisTemplate = redisTemplate;

        //设置好两个回调
        this.rabbitTemplate.setConfirmCallback((correlationData,  ack,   cause)->{
            if(!ack){
                //TODO 业务记录失败的消息； 本地消息表
                log.error("消息发送失败：消息id：{}"+correlationData.getId());
            }
        });

        this.rabbitTemplate.setReturnCallback((Message message,
                                               int replyCode,
                                               String replyText,
                                               String exchange,
                                               String routingKey)->{
            //TODO 业务记录失败的消息； 本地消息表
            log.error("消息队列存储失败：交换机：{}，路由键：{}，消息：{}，失败码：{}，失败原因：{}",exchange
            ,routingKey,message,replyCode,replyText);
        });

    }


    /**
     * 消息发送
     * @param exchange
     * @param routingKey
     * @param message
     */
    public void sendMsg(String exchange,String routingKey,Object message){
        //消息转json
        String jsonStr = JSONs.obj2JsonStr(message);
        //消息唯一id关联
        CorrelationData correlationData = new CorrelationData();
        correlationData.setId(UUID.randomUUID().toString());

        this.rabbitTemplate.convertAndSend(exchange,routingKey,jsonStr,correlationData);
    }


    public void retry(String msg, Long deliveryTag, Channel channel,Integer retryMax) throws IOException {
        //如果多次重试业务依然异常，需要将此消息保存到数据库【日志】，等待人工处理。
        //数据库连接断开。重试
        //计数器：超过5次失败就结束
        //同一个消息md5一样
        String msgMd5 = MD5.encrypt(msg);
        Long increment = redisTemplate.opsForValue().increment(RedisConst.RETRY_COUNT + msgMd5);
        //给一个过期时间。防止没有到5次的次数，redis数据没有删除，导致占用大量空间
        redisTemplate.expire(RedisConst.RETRY_COUNT + msgMd5,1, TimeUnit.DAYS);

        if(increment<=retryMax){
            log.info("消息处理失败：正在进行 第{}次 重试",increment);
            //睡眠多久后重试
            channel.basicReject(deliveryTag,true);
        }else {
            log.info("消息多次重试，业务依然失败，等待人工处理；消息内容：{}",msg);
            redisTemplate.delete(RedisConst.RETRY_COUNT + msgMd5);
            channel.basicAck(deliveryTag,false);
        }
    }
}
